package com.yanggu.flink.datastream_api.wordcount;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SocketStreamWordCount {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set(RestOptions.PORT, 8081);
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        environment
                .socketTextStream("localhost", 8888)
                .flatMap((value, out) -> {
                    for (String s : value.split(" ")) {
                        out.collect(s);
                    }
                }, Types.STRING)
                .map(value -> Tuple2.of(value, 1)).returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(tuple -> tuple.f0)
                .sum(1)
                .print("data print>>>");

        environment.execute("StreamWordCount");
    }

}
